package com.huoshan.im.server.service;

import com.alibaba.fastjson.JSON;
import com.huoshan.im.server.config.properties.SettingConfig;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import online.chazi.bootplus.commons.utils.MapUtils;
import online.chazi.netkit.messaging.Message;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Map;

/**
 * 说明：作者很懒，什么都没留下
 * Created by 火山九日 on 2020-04-27 13:37
 */
@Slf4j
@Service
@AllArgsConstructor
public class MessageKafkaService {

    private final SettingConfig settingConfig;
    private final KafkaTemplate<byte[], byte[]> kafkaTemplate;

    /**
     * @param fromUserId 允许为空
     * @param toUserId
     * @param message
     */
    public void unSendMessage(Long fromUserId, long toUserId, String clientType, Message message) {
        Map<String, Object> value = MapUtils.<String, Object>build()
                .add("toUserId", toUserId)
                .add("toClientType", clientType)
                .add("message", message);

        if (fromUserId != null) {
            value.put("fromUserId", fromUserId);
        }

        send(settingConfig.getKafka().getUnSendTopic(), JSON.toJSONString(value));
    }


    /**
     * @param fromUserId 允许为空
     * @param toUserId
     * @param message
     */
    public void send(Long fromUserId, long toUserId, String clientType, Message message) {
        Map<String, Object> value = MapUtils.<String, Object>build()
                .add("toUserId", toUserId)
                .add("message", message)
                .add("toClientType", clientType);
        if (fromUserId != null) {
            value.put("fromUserId", fromUserId);
        }
        send(settingConfig.getKafka().getSendTopic(), JSON.toJSONString(value));
    }

    /**
     * @param topic
     * @param data
     */
    public void send(String topic, String data) {

        log.debug("kafka发送消息:{}", data);

        try {
            ListenableFuture<SendResult<byte[], byte[]>> future = kafkaTemplate.send(new ProducerRecord<>(topic, data.getBytes()));
            future.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
                @Override
                public void onSuccess(SendResult<byte[], byte[]> sendResult) {
                    log.info("kafka发送消息成功：{}", data);
                }

                @Override
                public void onFailure(Throwable throwable) {
                    log.error("kafka发送消息失败：" + data, throwable);
                }
            });
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }

}
